# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: scanners\wvs\wvs_settings.py
"""WVS scan settings"""
from urllib.parse import urlsplit, urlunsplit
import xmltodict, re, os

__author__ = "r4w8173"


def normalize_path(path):
    return os.path.normpath(os.sep.join(re.split("\\\\|/", path)))


class WVSScanSettingsError(ValueError):
    """Invalid parameter combination"""

    def __init__(self, reason):
        super().__init__()
        self.reason = reason


class WVSScanSettings:
    """Represents scanning options to WVS. It converts between target configuration and settings.xml
    for launching WVS processes
    """

    def __init__(self, target_url, target_config=None):
        self._process = None
        self._logfile = None
        self.target_url = urlsplit(target_url)
        if self.target_url.scheme == "":
            self.target_url = urlsplit("http://" + target_url + "/")
        self.target_root = urlunsplit(
            (self.target_url.scheme, self.target_url.netloc, "/", "", "")
        )
        self.wvs_configuration = {"Settings": {}}
        self.read_only_settings_file = None
        if target_config is not None:
            self.configure(target_config)

    def _set_crawler_options(self, **kwargs):
        """Shortcut for settings crawler options from Settings/ScanSettings/Crawler path"""
        options = (
            self.wvs_configuration["Settings"]
            .setdefault("ScanSettings", {})
            .setdefault("Crawler", {})
        )
        for key in kwargs:
            options[key] = kwargs[key]

    def _set_scan_options(self, **kwargs):
        """Shortcut for settings scan options from Settings/ScanOptions path"""
        scan_options = self.wvs_configuration["Settings"].setdefault("ScanOptions", {})
        for key in kwargs:
            scan_options[key] = kwargs[key]

    def _set_scan_settings(self, **kwargs):
        """Shortcut for settings scan settings from Settings/ScanSettings path"""
        scan_settings = self.wvs_configuration["Settings"].setdefault(
            "ScanSettings", {}
        )
        for key in kwargs:
            if key in scan_settings:
                if isinstance(scan_settings[key], dict):
                    scan_settings[key].update(kwargs[key])
            else:
                scan_settings[key] = kwargs[key]

    def _add_http_authentication(self, url, username, password):
        """Add a new HTTP authentication credential set to the list of credentials"""
        http_authentication = (
            self.wvs_configuration["Settings"]
            .setdefault("Common", {})
            .setdefault("HTTPAuthentication", {"Credentials": {"Credential": []}})
        )
        http_authentication["Credentials"]["Credential"].append(
            {"@URL": url, "@Username": username, "@Password": password}
        )

    def _add_custom_http_header(self, name, value):
        """Add a new custom header to be sent while scanning"""
        custom_headers = (
            self.wvs_configuration["Settings"]
            .setdefault("ScanSettings", {})
            .setdefault("CustomHeaders", {})
            .setdefault("Header", [])
        )
        custom_headers.append(
            {"@enabled": "1", "@name": name.strip(), "@value": value.strip()}
        )

    def _add_custom_cookie(self, url, value):
        custom_cookies = (
            self.wvs_configuration["Settings"]
            .setdefault("ScanSettings", {})
            .setdefault("CustomCookies", {})
            .setdefault("Cookie", [])
        )
        custom_cookies.append({"@Url": url, "@CookieString": value})

    def _add_excluded_paths(self, paths, url):
        if len(paths) > 0:
            self.wvs_configuration["Settings"].setdefault("ScanOptions", {}).setdefault(
                "ExcludedPaths", {}
            ).setdefault("baseurl", []).append(
                {"@url": url, "path": [{"#text": path} for path in paths]}
            )

    def _add_client_certificate(self, url, certificate, password):
        """Add a client certificate to the list of client certificates"""
        certificates = (
            self.wvs_configuration["Settings"]
            .setdefault("Common", {})
            .setdefault("ClientCertificates", {"ClientCertificate": []})
        )
        certificates["ClientCertificate"].append(
            {
                "@CertFileName": normalize_path(certificate),
                "@Password": password,
                "@URL": url,
            }
        )

    def add_import(self, filename):
        """Add a file which should be imported by the scanner"""
        imports = (
            self.wvs_configuration["Settings"]
            .setdefault("ScanOptions", {})
            .setdefault("Import", {})
            .setdefault("File", [])
        )
        imports.append(normalize_path(filename))

    def configure(self, config, login_sequence=None, client_certificate=None):
        """Read a target configuration and update the settings xml dictionary accordingly"""
        if config is None:
            return
        if hasattr(config, "limit_crawler_scope"):
            if config.limit_crawler_scope:
                self._set_crawler_options(
                    FetchSubdirs="1", GetFirstOnly="0", RestrictToBaseFolder="1"
                )
            else:
                self._set_crawler_options(
                    FetchSubdirs="1", GetFirstOnly="0", RestrictToBaseFolder="0"
                )
            if hasattr(config, "case_sensitive"):
                if config.case_sensitive == "yes":
                    self._set_crawler_options(CaseInsensitiveURLs="0")
        if config.case_sensitive == "no":
            self._set_crawler_options(CaseInsensitiveURLs="1")
        else:
            if config.case_sensitive == "auto":
                self._set_scan_options(AutoCaseDetection="1")
            else:
                raise WVSScanSettingsError("Invalid case sensitivity config")
            if hasattr(config, "sensor"):
                if config.sensor:
                    if hasattr(config, "sensor_secret"):
                        if config.sensor_secret != "":
                            self._set_scan_settings(
                                Aspect={
                                    "Enabled": "1",
                                    "PasswordHash": config.sensor_secret,
                                }
                            )
                        raise WVSScanSettingsError("Expected sensor secret")
            if hasattr(config, "login"):
                pass
        if config.login.kind == "automatic":
            self._set_scan_options(
                Login={
                    "@kind": "automatic",
                    "LoginCredentials": {
                        "Username": config.login.credentials.username,
                        "Password": config.login.credentials.password,
                    },
                }
            )
        else:
            if config.login.kind == "sequence":
                if login_sequence is not None:
                    if login_sequence != "":
                        self._set_scan_options(
                            Login={"@kind": "sequence"},
                            LoginSequence=normalize_path(login_sequence),
                        )
                    raise WVSScanSettingsError("Expected login sequence file")
            else:
                if config.login.kind == "none":
                    pass
                else:
                    raise WVSScanSettingsError("Invalid login kind")
                if hasattr(config, "proxy"):
                    if config.proxy.get("enabled"):
                        if config.proxy.get("port") is None:
                            raise WVSScanSettingsError("Missing proxy port")
                        if config.proxy.get("address") is None:
                            raise WVSScanSettingsError("Missing proxy address")
                        if config.proxy.protocol == "http":
                            self._set_scan_settings(
                                Proxy={
                                    "HTTP": {
                                        "@Enabled": "1",
                                        "@Host": config.proxy.address,
                                        "@Port": str(config.proxy.port),
                                        "@Username": str(
                                            config.proxy.get("username", "")
                                        ),
                                        "@Password": str(
                                            config.proxy.get("password", "")
                                        ),
                                    }
                                }
                            )
                        else:
                            raise WVSScanSettingsError("Unrecognized proxy protocol")
                        if hasattr(config, "authentication"):
                            if config.authentication.enabled:
                                self._add_http_authentication(
                                    url=self.target_root,
                                    username=config.authentication.username,
                                    password=config.authentication.password,
                                )
                if hasattr(config, "scan_speed"):
                    if config.scan_speed == "sequential":
                        self._set_scan_settings(
                            HTTPGeneral={
                                "JobsPerMinute": "500",
                                "LimitConcurentRequests": "1",
                                "RequestDelay": "0",
                            }
                        )
                    else:
                        if config.scan_speed == "slow":
                            self._set_scan_settings(
                                HTTPGeneral={
                                    "JobsPerMinute": "1200",
                                    "LimitConcurentRequests": "2",
                                    "RequestDelay": "250",
                                }
                            )
        if config.scan_speed == "moderate":
            self._set_scan_settings(
                HTTPGeneral={
                    "JobsPerMinute": "3000",
                    "LimitConcurentRequests": "5",
                    "RequestDelay": "250",
                }
            )
        else:
            if config.scan_speed == "fast":
                self._set_scan_settings(
                    HTTPGeneral={
                        "JobsPerMinute": "0",
                        "LimitConcurentRequests": "10",
                        "RequestDelay": "0",
                    }
                )
            else:
                raise WVSScanSettingsError("Invalid scan speed settings")
            if hasattr(config, "technologies"):
                if len(config.technologies) > 0:
                    self._set_scan_options(
                        Technologies={
                            "Item": [
                                str(technology) for technology in config.technologies
                            ]
                        }
                    )
            if hasattr(config, "user_agent"):
                if len(config.user_agent) != "":
                    self._set_scan_settings(
                        HTTPGeneral={"DefaultAgent": config.user_agent}
                    )
            if client_certificate is not None:
                url = urlunsplit(
                    (
                        "https",
                        self.target_url.netloc
                        if self.target_url.scheme == "https"
                        else self.target_url.hostname,
                        "/",
                        "",
                        "",
                    )
                )
                try:
                    pwd = config.client_certificate_password
                except Exception:
                    pwd = ""

                self._add_client_certificate(url, client_certificate, pwd)
            for header in config.get("custom_headers", []):
                header = header.split(":", 1)
                if len(header) == 2:
                    self._add_custom_http_header(*header)
                else:
                    self._add_custom_http_header(header[0], "")

            for cookie in config.get("custom_cookies", []):
                self._add_custom_cookie(cookie.url, cookie.cookie)

            self._add_excluded_paths(config.get("excluded_paths", []), self.target_root)

    def setup_bxss(self, bxss_user_id, bxss_scan_id):
        self._set_values("Settings/Common/BlindXSS", Enabled="1", UserId=bxss_user_id)
        self._set_values("Settings/ScanOptions", BXSSScanId=bxss_scan_id)

    def load_from_file(self, filename):
        """Load a settings.xml file"""
        with open(filename, "r") as (settings_file):
            self.wvs_configuration = xmltodict.parse(settings_file.read())

    def load_read_only_settings(self, filename):
        """Loads a settings.xml file and fixes everything to be read only"""
        self.load_from_file(filename)
        self.read_only_settings_file = filename

    def unlock_read_only_settings(self):
        """Deletes the read only lock"""
        self.read_only_settings_file = None

    def add_allowed_host(self, url, config, client_certificate=None):
        """Adds an allowed host with some of it's settings"""
        split_url = urlsplit(url)
        allowed_hosts = (
            self.wvs_configuration["Settings"]
            .setdefault("ScanSettings", {})
            .setdefault("AllowedHosts", {"Host": []})
        )
        allowed_hosts["Host"].append(split_url.hostname)
        if client_certificate is not None:
            url = urlunsplit(
                (
                    "https",
                    split_url.netloc
                    if split_url.scheme == "https"
                    else split_url.hostname,
                    "/",
                    "",
                    "",
                )
            )
            try:
                pwd = config.client_certificate_password
            except Exception:
                pwd = ""

            self._add_client_certificate(url, client_certificate, pwd)
        if config is not None:
            root_url = urlunsplit((split_url.scheme, split_url.netloc, "/", "", ""))
            if hasattr(config, "authentication"):
                self._add_http_authentication(
                    url=root_url,
                    username=config.authentication.username,
                    password=config.authentication.password,
                )
            self._add_excluded_paths(config.get("excluded_paths", []), root_url)
            for header in config.get("custom_headers", []):
                header = header.split(":", 1)
                if len(header) == 2:
                    self._add_custom_http_header(*header)
                else:
                    self._add_custom_http_header(header[0], "")

            for cookie in config.get("custom_cookies", []):
                self._add_custom_cookie(cookie.url, cookie.cookie)

    def _set_values(self, path, **kwargs):
        last = self.wvs_configuration
        for node in path.split("/"):
            last = last.setdefault(node, {})

        for name, value in kwargs.items():
            last[name] = value

    def patch_for_quick_continuous(self):
        self._set_values(
            "Settings/ScanSettings/Parameters/ParameterManipulation/HTTPHeaders",
            Enabled="0",
        )
        self._set_values("Settings/ScanSettings/Parameters/ScanningMode", Mode="Quick")
        self._set_values("Settings/ScanSettings/WebKitOptions", SessionTimeout="10")
        self._set_values(
            "Settings/ScanSettings/Crawler",
            LinkDepth="15",
            MaxVariations="25",
            MaximumNumberOfPathSchemes="50",
            MaximumNumberOfPagesToCrawl="1000",
        )

    def set_interactive_mode(self, enabled):
        if enabled:
            self._set_scan_options(AllowInteractiveLogin="1")
        else:
            self._set_scan_options(AllowInteractiveLogin="0")

    def create_settings_xml_file(self, filename):
        """Un-parses and saves the updated settings.xml file"""
        if self.read_only_settings_file is not None:
            return self.read_only_settings_file
        with open(filename, "wb+") as (settings_file):
            xmltodict.unparse(
                self.wvs_configuration,
                output=settings_file,
                encoding="utf-8",
                pretty=True,
            )

    def get_target_url(self):
        """Returns the main target URL as it was specified at creation"""
        return urlunsplit(self.target_url)
